package com.sfx.mq.demo;

import com.sfx.mq.constants.MQConstant;
import com.sfx.mq.constants.TopicProperties;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.junit.Test;

import java.util.Arrays;
import java.util.Date;
import java.util.List;

/**
 * Created with IntelliJ IDEA.
 * Description: 发送批量消息
 * User: sfx
 * Date: 2023-10-27
 * Time: 15:39
 */
public class BatchMessageTest {

    @Test
    public void testProducer() throws Exception {
        // 创建生产者
        DefaultMQProducer producer = new DefaultMQProducer("test-producer-group");
        // 指定nameSrv
        producer.setNamesrvAddr(MQConstant.NAME_SRV);
        producer.start();
        // 发送消息
        Message message = new Message();
        message.setTopic(TopicProperties.BATCH_TOPIC);
        // 应用场景就是日志保存
        message.setBody("批量消息".getBytes());
        // 发送批量消息
        List<Message> msgs = Arrays.asList(
                new Message(TopicProperties.BATCH_TOPIC, "我是一组消息的A消息".getBytes()),
                new Message(TopicProperties.BATCH_TOPIC, "我是一组消息的B消息".getBytes()),
                new Message(TopicProperties.BATCH_TOPIC, "我是一组消息的C消息".getBytes())
        );
        producer.send(msgs);
        System.out.println(new Date());
        System.out.println("发送了批量消息");
        producer.shutdown();
    }

    @Test
    public void testConsumer() throws Exception {
        // 创建消费者
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("test-consumer-group");
        // 连接nameSrv
        consumer.setNamesrvAddr(MQConstant.NAME_SRV);
        // 订阅主题的消息,可以进行过滤
        consumer.subscribe(TopicProperties.BATCH_TOPIC, "*");
        // 创建一个消息监听器,监听队列中的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                System.out.println("我是消费者");
                String msg = new String(msgs.get(0).getBody());
                System.out.println("消费的消息为 : " + msg);
                System.out.println(new Date());
                System.out.println("消息上下文为 : " + consumeConcurrentlyContext);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });

        // 开启消费者
        consumer.start();
        // 挂起JVM,不要停
        System.in.read();
    }
}